package com.alison.datastream.chapter6_timeAndWaterMark.watermark;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author alison
 * @Date 2024/4/4 22:09
 * @Version 1.0
 * @Description
 */
public class W4_WaterMark1Test {
    /*
    nc -lp 12345
    数据，
    h1,1553503181000
    h1,1553503182000
    h1,1553503183000
    h1,1553503184000
    h1,1553503185000
    h1,1553503186000
    h1,1553503187000
    h1,1553503188000
    h1,1553503189000
    h1,1553503190000
    h1,1553503187000
     */

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setMaxParallelism(1).setParallelism(1);
//        env.getConfig().setAutoWatermarkInterval(5000L);//tigger fun of onPeriodicEmit
//        SingleOutputStreamOperator<String> dataStream = env.socketTextStream("192.168.56.101", 12345);
        DataStreamSource<String> dataStream = env.addSource(new MySourceForLaten());
        DataStream source = dataStream.map((ele) -> {
            String[] split = ele.split(",");
            return Tuple2.of(split[0], Long.parseLong(split[1].trim()));
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));
        SingleOutputStreamOperator<Tuple2<String, Long>> input = source.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {
            @Override
            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Tuple2<String, Long>>() {
                    private long maxTimestamp = 0L;
                    private long delay = 5000; //延迟3s
                    private long currentWatermark;// 当前水位线

                    @Override
                    public void onEvent(Tuple2<String, Long> event, long eventTimestamp, WatermarkOutput output) {
                        maxTimestamp = Math.max(eventTimestamp, maxTimestamp);
                        System.out.println("key: " + event.f0 + ", value:" + event.f1 + ",前一条水位线： " + currentWatermark + ", 最大的ts: " + maxTimestamp);
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        currentWatermark = maxTimestamp - delay;
                        System.out.println("onPeriodicEmit--" + currentWatermark);
                        output.emitWatermark(new Watermark(currentWatermark));
                    }
                };
            }
        }.withTimestampAssigner((new SerializableTimestampAssigner<Tuple2<String, Long>>() {
            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                System.out.println("extractTimestamp--" + element.f1);
                return element.f1;
            }
        })));
        // allowedLateness 允许延迟5s, 每个元素的eventTime跟watermark的时间比较，在5s的延迟内的元素，来一次，触发一下windows类的计算，超过这些数据，丢弃
        // 没有配置延迟水印，发送延迟的数据直接丢弃
        input.keyBy(item -> item.f0).window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .allowedLateness(Duration.ofSeconds(5))//允许延迟5s
                .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>() {
                    public void process(String key, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Tuple2<String, Long>> out) throws Exception {
                        long count = 0;
                        StringBuilder tsList = new StringBuilder();
                        for (Tuple2<String, Long> element : elements) {
                            count += 1;
                            tsList.append(element.f1 + " ");
                        }
                        System.out.println("start: " + tsList);
                        out.collect(Tuple2.of(key, count));
                    }
                }).print();
        env.execute();
    }


}
